use crate::sip2::{
	dialog::DialogEvent,
	transport::{TransportLayerPacket, TransportProtocol},
};
use base_data::outside::switch::OutSideCommand;
use bytes::BytesMut;
use std::{net::SocketAddr, sync::Arc};
use tokio::{
	net::UdpSocket,
	sync::{broadcast, mpsc},
};
use tools::{global_defin::tokio_runtime::get_rt, ResultEx};
use tracing::{error, info, span, trace, Instrument, Level};

///udp的监听器,此方法每执行一次将产生一个监听器.并将收到的消息向队列发送.
/// 目前收到的单个消息不能超过10k.超过部分有可能被丢弃.
/// recv_switch接收全局所有的针对对外接口的消息.这样可以减少复杂性,但需要根据自己的地址才进行处理
pub async fn listener_udp(
	addr: SocketAddr,
	sender: mpsc::Sender<DialogEvent>,
	recv_switch: broadcast::Receiver<OutSideCommand>,
) -> ResultEx<Arc<UdpSocket>> {
	let add = addr.to_string();
	let span = span!(Level::DEBUG, "udps", add);

	info!("开始监听:{addr:?}");

	//由于需要tcp和udp同时监听同一个地址和端口.所以将收到的消息转发到同一个处理器中.
	let sock = match UdpSocket::bind(addr).await {
		Ok(s) => Arc::new(s),
		Err(e) => {
			error!("绑定异常: {:?}", e);
			return Err(e.into());
		}
	};

	get_rt().spawn(start_lister(sock.clone(), sender, recv_switch).instrument(span));

	Ok(sock)
}

///开始监听并接收来到的数据.
async fn start_lister(
	sock: Arc<UdpSocket>,
	sender: mpsc::Sender<DialogEvent>,
	mut _recv_switch: broadcast::Receiver<OutSideCommand>,
) {
	//每个监听使用最大缓存20M
	let cap = 20 * 1024 * 1024;
	let mut buf = BytesMut::with_capacity(cap);
	// let local_addr = sock.local_addr().unwrap();
	loop {
		match sock.recv_buf_from(&mut buf).await {
			Ok((len, remote_addr)) => {
				let data = buf.split_to(len).freeze();
				trace!("接收到数据:\n{}", unsafe {
					std::str::from_utf8_unchecked(&data)
				});
				
				//判断一下如果过短,直接跳过.
				if len < 20 {
					continue;
				}

				if let Err(e) = sender
					.send(DialogEvent::TransportLayerPacket(TransportLayerPacket {
						data,
						protocol: TransportProtocol::Udp(sock.clone(), remote_addr),
					}))
					.await
				{
					error!("listener_udp()发送数据包异常: {:?}", e);
				}

				if buf.capacity() < 5 * 1024 {
					buf = BytesMut::with_capacity(cap);
				}
			}
			Err(e) => {
				error!("listener_udp()接收数据包异常: {:?}", e);
			}
		}
	}
	// tokio::select! {
	// 	biased;
	// 	msg = sock.recv_buf_from(&mut buf) => {
	// 		match msg {
	// 			Ok((len, remote_addr)) => {
	// 				//判断一下如果过短,直接跳过.
	// 				if len < 20 {
	// 					buf.clear();
	// 					continue;
	// 				}
	// 				let data = buf.split_to(len).freeze();
	// 				trace!("接收到数据:\n{}", unsafe {
	// 					std::str::from_utf8_unchecked(&data)
	// 				});
	// 				if let Err(e) = sender
	// 					.send(DialogEvent::TransportLayerPacket(TransportLayerPacket {
	// 						data,
	// 						// local_addr,
	// 						// remote_addr:remote_addr.clone(),
	// 						protocol:TransportProtocol::Udp(sock.clone(),remote_addr),
	// 					}))
	// 					.await
	// 				{
	// 					error!("listener_udp()发送数据包异常: {:?}", e);
	// 				}
	// 			}
	// 			Err(e) => {
	// 				error!("listener_udp()接收数据包异常: {:?}", e);
	// 			}
	// 		}

	// 		if buf.capacity() < 5 * 1024 {
	// 			buf = BytesMut::with_capacity(cap);
	// 		}
	// 	},
	// 	msg = recv_switch.recv() => {
	// 		match msg {
	// 			Ok(msg) => {
	// 				if msg.addr == local_addr && msg.switch == SwitchOutSide::Close {
	// 					info!("{}收到关闭指令,退出当前监听",msg.addr);
	// 					return;
	// 				}
	// 			}
	// 			Err(e) => {
	// 				error!("listener_udp()接收命令异常: {:?}", e);
	// 				// return;
	// 			}
	// 		}
	// 	},
	// }
}
